package com.arch.flink.biz.orderdemo;

import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.io.BufferedReader;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.net.URI;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

/**
 * @author pizhihui
 * @date 2020-12-12
 */
public class OrderInfoSource implements SourceFunction<String> {


    private Random random = new Random();

    private String filename;

    public OrderInfoSource(String filename){
        this.filename = filename;
    }


    @Override
    public void run(SourceContext<String> ctx) throws Exception {
        List<String> allLines = Files.readAllLines(Paths.get(filename));
        for (String allLine : allLines) {
            TimeUnit.MILLISECONDS.sleep(random.nextInt(500));
            ctx.collect(allLine);
        }
    }

    @Override
    public void cancel() {

    }


}
